package com.alison.sparkstream.source

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.util.Random
import scala.collection.mutable

object E3_stream_rdd_queue {

  """
    |RDD 队列流是指通过将 RDD 队列作为数据源实时处理数据的一种方式。这种方法适用于调试和测试的目的，可以手动将数据推送到 Spark Streaming 中进行处理。
    |在 Spark Streaming 中，可以通过 StreamingContext.queueStream(queueOfRDDs) 来创建一个 RDD 队列流。
    |
    |下面是一个简单的示例代码，演示如何使用 RDD 队列流处理手动创建的 RDD 数据。我们会定时向 RDD 队列中放入新的 RDD，
    |Spark Streaming 则通过 RDD 队列流进行词频率统计。同时为了增加一些随机性，我们在生成每个 RDD 时加入一些随机的数据。
    |""".stripMargin

  def main(args: Array[String]): Unit = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ncExample")
    // 初始化 StreamingContext，设置微批处理的时间间隔为 3 秒
    val streamingContext = new StreamingContext(sparkConf, Seconds(3))
    // 创建RDD队列流
    val rddQueue = new mutable.Queue[RDD[String]]()
    val lineStreams = streamingContext.queueStream(rddQueue)
    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = lineStreams.flatMap(_.split(" "))
    // 将单词映射成元组（word,1）
    val wordAndOneStreams = wordStreams.map((_, 1))
    // 将相同的单词次数做统计
    val wordAndCountStreams = wordAndOneStreams.reduceByKey(_ + _)
    // 打印结果
    wordAndCountStreams.print()
    // 启动 StreamingContext
    streamingContext.start()

    for (i <- 1 to 5) {
      val randomText = generateRadomText()
      val rdd = streamingContext.sparkContext.parallelize(Seq(randomText))
      rddQueue += rdd
      Thread.sleep(5_000) // 每5秒放入一个新的RDD
    }
    // 等待应用程序终止（遇错停止，否则不断工作）
    streamingContext.awaitTermination()
  }

  def generateRadomText() = {
    val words = List("hello", "world", "spark", "streaming", "flink")
    val random = new Random()
    val randomWords = (1 to 10).map(_ => words(random.nextInt(words.length)))
    randomWords.mkString(" ")
  }
}
